package com.tianlh.zookeeper;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

/**
 * @author 黑子
 * @Date 2018/1/9 20:12
 */
public class ZkClient {

// -----------------------------------------------------------------创建zk客户端

    static RetryPolicy retry = new ExponentialBackoffRetry(1000, 3);
    // 为了实现不同的ZooKeeper业务之间的隔离，往往会为每个业务分配一个独立的命名空间，即指定一个ZooKeeper根路径
    // 该客户端对ZooKeeper上数据节点的任何操作，都是基于"/base"该相对目录进行的
    static CuratorFramework client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181")
        .namespace("base").sessionTimeoutMs(5000).retryPolicy(retry).build();

    public static void main(String[] args) throws Exception {
// -----------------------------------------------------------------启动客户端

        client.start();

// -----------------------------------------------------------------创建节点

        // 创建节点,如果没有设置节点属性，那么默认创建的是持久节点
        client.create().forPath("/heizi", "zuishuai".getBytes());
        // 创建临时节点
        client.create().withMode(CreateMode.EPHEMERAL).forPath("/dayue", "woaini".getBytes());
        // 开发时经常会碰到NoNodeException异常，其中一个可能的原因就是试图对一个不存在的父节点创建子节点。
        // 因此不得不在每次创建节点之前，都判断－下该父节点是否存在，这很麻烦，利用下面的函数解决了该问题。
        // 创建一个临时节点，并自动逆归创建父节点,，注意：在ZooKeeper中规定了所有非叶子节点、必须为持久节点
        client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath("/zk/test", "test".getBytes());

// -----------------------------------------------------------------节点监听
        final NodeCache nodeCache = new NodeCache(client, "/dayue", false);
        // 默认是false,如果设置为true,那么NodeCache在第一次启动的时候就会立刻从ZooKeeper上读取对应节点的数据内容,并保存在Cache中。
        nodeCache.start(false);
        nodeCache.getListenable().addListener(new NodeCacheListener() {
            // 节点内容改变，节点创建和删除都会触发
            @Override
            public void nodeChanged() throws Exception {
                System.out.println("更新后的数据："+ new String(nodeCache.getCurrentData().getData()));
            }
        });

// -----------------------------------------------------------------子节点监听(只能监听子一级)
        final PathChildrenCache childrenCache = new PathChildrenCache(client, "/zk", false);
        // 开发者也可以通过构造一个专门的线程池,来处理事件通知
        //PathChildrenCache(CuratorFramework client,String path,boolean cacheData,boolean datalsCompressed, final ExecutorService executor);
        childrenCache.start(StartMode.POST_INITIALIZED_EVENT);
        childrenCache.getListenable().addListener(new PathChildrenCacheListener() {
            // 新增子节点CHILD_ADDED、子节点数据变更CHILD_ UPDATED和子节点删除CHILD_REMOVED时会触发，但是当前节点的变更不会触发
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event)
                throws Exception {
                switch(event.getType()){
                    case CHILD_ADDED:
                        System.out.println("CHILD_ADDED:" + event.getData().getPath());
                        break;
                    case CHILD_UPDATED:
                        System.out.println("CHILD_UPDATED:" + event.getData().getPath());
                        break;
                    case CHILD_REMOVED:
                        System.out.println("CHILD_REMOVED:" + event.getData().getPath());
                        break;
                    default:
                        break;
                }
            }
        });


// -----------------------------------------------------------------读取节点数据

        // 查询一个节点的内容
        client.getData().forPath("/zk/test");
        // 查询一个节点的数据内容,并返回stat,返回的stat内容存储在传进去的stat对象中
        Stat stat = new Stat();
        client.getData().storingStatIn(stat).forPath("/dayue");

// -----------------------------------------------------------------更新节点数据

        // 更新一个节点的数据内容，并返回一个stat对象
        Stat dayueStat = client.setData().forPath("/dayue", "wohenni".getBytes());
        // 更新一个节点的数据内容，强制指定版本进行更新,withVersion接口就是用来实现CAS的,version
        // 通常是从一个旧的stat 对象中获取到的,使用过期的stat变量进行更新操作将抛出异常
        Stat stat2 = client.setData().withVersion(dayueStat.getVersion()).forPath("/zk/test", "newvalue".getBytes());

// -----------------------------------------------------------------删除节点
        // 删除一个节点，并递归删除其所有子节点
        client.delete().deletingChildrenIfNeeded().forPath("/zk");
        // 删除一个节点，强制指定版本进行删除
        client.delete().withVersion(0).forPath("/heizi");
        // 删除一个节点，强制保证删除,只要客户端会话有效，那么Curator会在后台持续进行删除操作，直到节点删除成功。
        client.delete().guaranteed().forPath("/heizi");

        Thread.sleep(5000);

    }
}
